import com.tsj.message.Producer;
import com.tsj.message.config.KafkaProperties;
import org.apache.kafka.common.errors.TimeoutException;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class ProducerStarter {

    public static void main(String[] args) throws InterruptedException {
        boolean isAsync = args.length == 0 || !args[0].trim().equalsIgnoreCase("sync");
        CountDownLatch latch = new CountDownLatch(1);

        for (int id = 0; id < 10; id++) {
            Producer producerThread = new Producer("DemoProducer_" + id, KafkaProperties.TOPIC, isAsync, null,
                    false, 10, -1,
                    latch, "user_" + id);
            producerThread.start();
            // 等待执行完成后，再次执行，保证一组的offset是连续的
            producerThread.join();
        }

        if (!latch.await(30, TimeUnit.SECONDS)) {
            throw new TimeoutException("Timeout after 30 seconds waiting for demo producer to finish");
        }
        System.out.println("demo producer to finish!");
    }
}
